// Copyright (c) 2022 Presto Labs Pte. Ltd.
// Author: gekim

#include "coin2/exchange/prex_base/feed_v1/processor.h"

#include <string>
#include <utility>
#include <boost/algorithm/string/replace.hpp>

#include "coin2/base/conversion.h"
#include "coin2/base/string_util.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/prex_base/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::prex_base::feed_v1::impl {

void PrexBaseFeedProcessor::OnTopicRecord(const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    throw std::runtime_error(
        std::string("JSON parse error: ") + GetParseError_En(document.GetParseError()));
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (document.HasMember("response") && document["response"].HasMember("request")) {
    if (document["response"]["request"]["op"] == "subscribe") {
      upd_mgr_.onEndPacket(packet);
      return;
    }
  }

  if (!document.HasMember("topic") || !document.HasMember("data")) {
    LOG(INFO) << "[Prex Feed Processor] meta msg for " << packet.topic_name << ": "
              << packet.payload.data();
    upd_mgr_.onEndPacket(packet);
    return;
  }

  const std::string topic = document["topic"].GetString();

  const auto& channel = PrexChannel::Parse(topic);
  const auto& native_symbol = channel.symbol;

  if (!CheckChannelFilter(channel.type)) {
    upd_mgr_.onEndPacket(packet);
    return;
  }

  if (channel.type == "market/orderBook") {
    auto ts = packet.timestamp;
    auto* book = upd_mgr_.GetBookMutable(native_symbol);
    if (book != nullptr){
      bool generated = false;
      generated = PrexBaseFeedUpdateParser::ParseBook(book, ts, document["data"].GetArray());
      if (generated) {
        upd_mgr_.PublishBook();
      }
    }
  } else if (channel.type == "market/trade") {
    if (document["data"].Size() > 1) {
      LOG(ERROR) << "More than one type in a single ws feed message, reimplement!\n"
                 << packet.payload.data();
    }
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }

    Trade* agg_trade = upd_mgr_.GetTradeMutable();
    agg_trade->Clear();
    Trade trade{};

    for (auto& trade_json : document["data"][0]["rows"].GetArray()) {
      PrexBaseFeedUpdateParser::ParseTrade(trade_json, &trade);
      if (TryAggregateTrades(agg_trade, trade)) {
        continue;
      }
      if (agg_trade->fill_qty > 0) {
        upd_mgr_.PublishTrade(native_symbol, true);
      }
      *agg_trade = trade;
    }
    if (agg_trade->fill_qty > 0) {
      upd_mgr_.PublishTrade(native_symbol, false);
    }
  } else if (channel.type == "market/instrument") {
    for (auto& data_list : document["data"].GetArray()) {
      if (data_list["actionType"] == "UPDATE"){
        for (auto& instrument : data_list["rows"].GetArray()) {
          const std::string& native_symbol = instrument["symbol"].GetString();
          std::string native_to_check_group = native_symbol;
          if (StringEndsWith(native_symbol, "IDX")) {
            native_to_check_group = boost::replace_all_copy(native_symbol, "IDX", "PERP");
          }
          if (native_to_topics_.count(native_to_check_group) == 0 ||
              native_to_topics_[native_to_check_group].count(packet.topic_name) == 0) {
            continue;
          }
          if (upd_mgr_.is_symbol_registered(native_symbol)) {
            auto* mark_price = upd_mgr_.GetMarkPriceMutable();
            auto* funding_rate = upd_mgr_.GetFundingRateMutable();
            if (PrexBaseFeedUpdateParser::ParseMarkPrice(
                instrument, mark_price, packet.timestamp)) {
              upd_mgr_.PublishMarkPrice(native_symbol);
            }
            if (PrexBaseFeedUpdateParser::ParseFundingRate(
                instrument, funding_rate, packet.timestamp)) {
              upd_mgr_.PublishFundingRate(native_symbol);
            }
          } else if (StringEndsWith(native_symbol, "IDX") && instrument.HasMember("lastPrice")) {
            const std::string& inherent_native_symbol = boost::replace_all_copy(
                native_symbol, "IDX", "PERP");
            if (upd_mgr_.is_symbol_registered(inherent_native_symbol)) {
              auto* index = upd_mgr_.GetIndexMutable();
              PrexBaseFeedUpdateParser::ParseIndex(
                  instrument, index, packet.timestamp);
              upd_mgr_.PublishIndex(inherent_native_symbol);
            }
          }
        }
      }
      if (data_list["actionType"] == "SNAPSHOT"){
        for (auto& instrument : data_list["rows"].GetArray()) {
          const std::string& native_symbol = instrument["symbol"].GetString();
          if (StringEndsWith(native_symbol, "PERP") &&
              upd_mgr_.is_symbol_registered(native_symbol)) {
            auto* funding_rate = upd_mgr_.GetFundingRateMutable();
            if (PrexBaseFeedUpdateParser::ParseFundingRate(
                instrument, funding_rate, packet.timestamp)) {
              upd_mgr_.PublishFundingRate(native_symbol);
            }
          }
        }
      }
    }
  } else {
    THROW() << "[Prex Feed Processor] unkown topic: " << packet.payload;
  }

  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::prex_base::feed_v1::impl
